package com.xixi.canal.client.starter.loader;

import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.xixi.canal.client.DataWriter;
import com.xixi.canal.client.starter.config.InstanceConfig;

import java.util.List;

public class CanalClientWorker extends AbstractCanalClientWorker{

    private static final int BATCH_SIZE = 1024;

    public CanalClientWorker(InstanceConfig instanceConfig,
                             String destination,
                             List<DataWriter> writers,
                             CanalServerWithEmbedded embeddedServer) {
        super(writers);
        this.destination = destination;
        this.instanceConfig = instanceConfig;
        this.embeddedServer = embeddedServer;
        clientIdentity = new ClientIdentity(destination, (short) 1001);
    }


    @Override
    protected void process() {
        while (!running) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
            }
        }
        int retry = instanceConfig.getRetries() == null || instanceConfig.getRetries() == 0 ? 1 : instanceConfig.getRetries();
        if (retry == -1) {
            // 重试次数-1代表异常时一直阻塞重试
            retry = Integer.MAX_VALUE;
        }
        Integer batchSize = instanceConfig.getBatchSize();
        if (batchSize == null) {
            batchSize = BATCH_SIZE;
        }

        while (running) {

            try {
                syncSwitch.get(destination);
                logger.info("=============> Start to connect destination: {} <=============", this.destination);
//                connector.connect();
                logger.info("=============> Start to subscribe destination: {} <=============", this.destination);
//                connector.subscribe();
//                unsubscribe();
                subscribe();
                logger.info("=============> Subscribe destination: {} succeed <=============", this.destination);

                while (running) {

                    boolean state = syncSwitch.getState(destination);
                    if (!state) break;

                    if (!running) break;

                    for (int i = 0; i < retry; i++) {
                        if (!running) break;

                        Message message = getWithoutAck(batchSize);
                        long batchId = message.getId();
                        try {
                            int size = message.getEntries().size();

                            if (batchId == -1 || size == 0) {

                            } else {
                                if (logger.isDebugEnabled()) {
                                    logger.debug("destination: {} batchId: {} batchSize: {} ",
                                            destination,
                                            batchId,
                                            size);
                                }
                                long begin = System.currentTimeMillis();
                                writeMessage(message);
                                if (logger.isDebugEnabled()) {
                                    logger.debug("destination: {} batchId: {} elapsed time: {} ms",
                                            destination,
                                            batchId,
                                            System.currentTimeMillis() - begin);
                                }
                            }

                            if (batchId != -1) {
                                ack(batchId);
                            }
                            break;
                        } catch (Exception e) {
                            if (i != retry - 1) {
                                rollback(batchId);
                                logger.error("Destination: {} batchId: {} sync error and rollback, execute times: {},{}",
                                        destination,
                                        batchId,
                                        (i + 1),
                                        e);
                            } else {
                                ack(batchId);
                                logger.error("Destination: {} batchId: {} sync error and but ack:{}",
                                        destination,
                                        batchId,
                                        e);
                            }
                            Thread.sleep(500);
                        }
                        message = null;
                    }
                }
            } catch (Throwable e) {
                logger.error("process error,{}", e);
            } finally {
//                unsubscribe();
                logger.info("=============> Disconnect destination: {} <=============", this.destination);
            }


            if (running) { // is reconnect
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // ignore
                }
            }
        }

    }

    @Override
    protected void destroy() {
//        if (embeddedServer.isStart(destination)) {
//            embeddedServer.stop(destination);
//        }
    }
}
